#include "mprpcprovider.h"

/**
 * service_name => service描述（service* 记录服务对象）
 * method_name  => method方法对象
*/
// 这里是框架提供给外部使用的，可以发布rpc方法的函数接口
void RpcProvider::NotifyService(google::protobuf::Service* service)
{
    ServiceInfo serviceInfo;

    // 获取服务对象的描述信息
    const google::protobuf::ServiceDescriptor* pserviceDesc = service->GetDescriptor();
    // 获取服务的名字，也就是proto文件中定义的service的名字
    std::string serviceName = pserviceDesc->name();
    // 获取服务对象service的方法数量
    int methodCount = pserviceDesc->method_count();

    // std::cout << "serviceName: " << serviceName << std::endl;
    LOG_INFO("serviceName:%s", serviceName.c_str());

    for(int i = 0; i < methodCount; ++i)
    {
        // 获取了服务对象指定下标的服务方法的描述（抽象描述）
        const google::protobuf::MethodDescriptor* pmethodDesc = pserviceDesc->method(i);
        std::string methodName = pmethodDesc->name(); // 也就是proto文件中定义的rpc方法的名字
        // std::cout << "methodName: " << methodName << std::endl;
        LOG_INFO("methodName:%s", methodName.c_str());
        serviceInfo._methodMap.insert({methodName, pmethodDesc});
    }

    serviceInfo._service = service;
    _serviceMap.insert({serviceName, serviceInfo});
}

// 启动rpc服务节点，开始提供rpc远程网络调用服务
void RpcProvider::Run()
{
    std::string ip = MprpcApplication::getInstance().getConfig().Load("rpcserverip");
    uint16_t port = atoi(MprpcApplication::getInstance().getConfig().Load("rpcserverport").c_str());
    muduo::net::InetAddress address(ip, port);

    // 创建TcpServer对象
    muduo::net::TcpServer server(&_eventLoop, address, "RpcProvider");
    // 绑定连接回调和消息读写回调方法(分离了网络代码和业务代码)
    server.setConnectionCallback(std::bind(&RpcProvider::onConnection, this, std::placeholders::_1));
    server.setMessageCallback(std::bind(&RpcProvider::onMessage, this, std::placeholders::_1, 
            std::placeholders::_2, std::placeholders::_3));

    // 设置muduo库的线程数量
    server.setThreadNum(3);

    // 把当前rpc节点上要发布的服务全部注册到zookeeper上，让rpc client可以从zookeeper上发现服务
    // session timeout超时时间是30秒，zkclient的网络I/O线程会以三分之一的timeout超时时间发送心跳消息
    ZkClient zkCli;
    zkCli.Start(); // 连接zookeeper server
    // serviceName为永久性节点  methodName为临时性节点
    for(auto& servicePair : _serviceMap)
    {
        /****** /serviceName（服务节点 /UserServiceRpc） ******/   
        std::string servicePath = "/" + servicePair.first;
        zkCli.Create(servicePath.c_str(), nullptr, 0);
        for(auto& methodPair : servicePair.second._methodMap)
        {
            /********  /serviceName/methodName（方法节点 /UserServicePrc/Login） *********/ 
            // 存储的是当前这个rpc服务节点主机的ip地址和端口号
            std::string methodPath = servicePath + "/" + methodPair.first;
            char methodPathData[128] = {0};
            sprintf(methodPathData, "%s:%d", ip.c_str(), port);
            // ZOO_EPHEMERAL表示znode是一个临时性节点
            zkCli.Create(methodPath.c_str(), methodPathData, strlen(methodPathData), ZOO_EPHEMERAL); 
        }
    }

    std::cout << "RpcProvider start service at ip: " << ip << " port: " << port << std::endl;

    // 启动网络服务
    server.start();
    _eventLoop.loop();
}

// 新的socket连接回调
void RpcProvider::onConnection(const muduo::net::TcpConnectionPtr& conn)
{
    // rpc请求是短连接，请求完就可以关闭连接
    if(!conn->connected())
    {
        conn->shutdown(); // 关闭文件描述符
    }
}

/**
 * 在框架内部，RpcProvider和RpcConsumer需要协商好用于通信的protobuf数据类型
 * serviceName methodName args 
 * 那么就需要定义proto的message类型，进行数据头的序列化和反序列化
 * headerSize(4个字节) + headerStr(serviceName methodeName argsSize) + argsStr
 * 如：18UserServiceLogin15zhang san123456
*/
// 已建立连接用户的读写事件回调
// 如果远程有一个rpc服务的调用请求，那么onMessage方法就会被调用
void RpcProvider::onMessage(const muduo::net::TcpConnectionPtr& conn, 
                            muduo::net::Buffer* buf, 
                            muduo::Timestamp time)
{
    // 网络上接收到远程rpc调用请求的字符流 Login args
    std::string recvBuf = buf->retrieveAllAsString();

    // 从字符流中读取前4个字节的内容
    uint32_t headerSize = 0;
    recvBuf.copy((char*)&headerSize, 4, 0);
    
    // 根据headSize大小来读取数据头的原始字符流
    std::string rpcHeaderStr = recvBuf.substr(4, headerSize);
    // 反序列化数据，得到rpc请求的详细信息
    mprpc::RpcHeader rpcHeader;
    std::string serviceName;
    std::string methodName;
    uint32_t argsSize;
    if(rpcHeader.ParseFromString(rpcHeaderStr))
    {
        // 数据头反序列化成功
        serviceName = rpcHeader.servicename();
        methodName = rpcHeader.methodname();
        argsSize = rpcHeader.argssize();
    }
    else
    {
        // 数据头反序列化失败
        std::cout << "rpcHeaderStr: " << rpcHeaderStr << " parse error!" << std::endl;
        return;
    }

    // 获取rpc方法参数的字符流数据
    std::string argsStr = recvBuf.substr(4 + headerSize, argsSize);

    // 打印调试信息
    /*
    std::cout << "==========================================" << std::endl;
    std::cout << "headerSize: " << headerSize << std::endl;
    std::cout << "rpcHeaderStr: " << rpcHeaderStr << std::endl;
    std::cout << "serviceName: " << serviceName << std::endl;
    std::cout << "methodName: " << methodName << std::endl;
    std::cout << "argsSize: " << argsSize << std::endl;
    std::cout << "argsStr: " << argsStr << std::endl;
    std::cout << "==========================================" << std::endl;
    */

    // 获取service对象和method对象
    auto it1 = _serviceMap.find(serviceName);
    if(it1 == _serviceMap.end())
    {
        std::cout << serviceName << " does not exist!" << std::endl;
        return;
    }

    auto it2 = it1->second._methodMap.find(methodName);
    if(it2 == it1->second._methodMap.end())
    {
        std::cout << serviceName << ":" << methodName << " does not exist!" << std::endl;
        return; 
    }
    
    // 获取service对象（该对象实际是指向UserService的）
    google::protobuf::Service* service = it1->second._service; 
    // 获取method对象（Login）
    const google::protobuf::MethodDescriptor* method = it2->second; 

    // 生成rpc方法调用的请求Request和响应Response参数
    google::protobuf::Message* request = service->GetRequestPrototype(method).New();
    if(!request->ParseFromString(argsStr))
    {
        std::cout << "argsStr: " << argsStr << " parse error!" << std::endl;
        return;
    }
    google::protobuf::Message* response = service->GetResponsePrototype(method).New();

    // 给下面的method方法的调用，绑定一个Closure的回调函数
    google::protobuf::Closure* done = google::protobuf::NewCallback<RpcProvider, 
                               const muduo::net::TcpConnectionPtr&, google::protobuf::Message*>
                               (this, &RpcProvider::sendRpcResponse, conn, response);

    // 在框架上，根据远程rpc请求，调用当前rpc节点上发布的方法
    // new UserService(controller, request, response, done)
    service->CallMethod(method, nullptr, request, response, done); 
}

// Closure的回调操作，用于序列化rpc的响应和网络发送
void RpcProvider::sendRpcResponse(const muduo::net::TcpConnectionPtr& conn, google::protobuf::Message* response)
{
    // response进行序列化
    std::string responseStr;
    if(response->SerializeToString(&responseStr))
    {
        // 序列化成功，通过网络将rpc方法执行的结果发送给rpc的调用方
        conn->send(responseStr);
    }
    else
    {
        std::cout << "serialize response error!" << std::endl;
    }
    conn->shutdown(); // 模拟http的短连接服务，由RpcProvider主动断开连接
}